package com.winning.xffw.service.service.impl;


import com.winning.xffw.core.utils.CsvUtils;
import com.winning.xffw.core.utils.UUIDUtils;
import com.winning.xffw.service.entity.dto.BduKafkaMsg;
import com.winning.xffw.service.mapper.BduKafkaMsgMapper;
import com.winning.xffw.service.service.IBduKafkaMsgService;
import com.winning.xffw.service.service.IDataBaseService;
import lombok.extern.slf4j.Slf4j;
import org.assertj.core.util.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.File;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * @Author: shipengfei
 * @Description:
 * @Date: 2019-05-31 10:29
 */
@Service
@Slf4j
public class BduKafkaMsgServiceImpl implements IBduKafkaMsgService {

    @Autowired
    private BduKafkaMsgMapper bduKafkaMsgMapper;

    @Autowired
    private IDataBaseService dataBaseService;


    /**
     * 数据消息表保存
     *
     * @param msg
     */
    @Transactional
    @Override
    public void saveKafkaMsg(BduKafkaMsg msg) {
        bduKafkaMsgMapper.insertTableData(msg);
    }

    /**
     * 查待处理的表
     *
     * @return
     */
    @Override
    public List<String> queryTableList() {
        return bduKafkaMsgMapper.queryTableList();
    }

    /**
     * 根据表名获取待处理记录
     *
     * @param tableName
     * @return
     */
    @Override
    public List<BduKafkaMsg> queryDataMsgByTable(String tableName) {
        List<BduKafkaMsg> msgList = bduKafkaMsgMapper.queryDataMsgByTable(tableName);
        // 通过copyOut获取记录
//        List<BduKafkaMsg> msgList = findDataMsgByTable(tableName);
        return msgList;
    }


    /**
     * 按照表名获取待处理的记录
     *
     * @param tableName
     * @return
     */
    private List<BduKafkaMsg> findDataMsgByTable(String tableName) {
        String basePath = System.getProperty("user.dir") + File.separator + "tempData" + File.separator
                + new SimpleDateFormat("yyyyMMdd").format(new Date()) + File.separator
                + UUIDUtils.getUUID() + File.separator;
        new File(basePath).mkdirs();
        String filename = basePath + tableName + "_msg.csv";
        StringBuilder sqlStr = new StringBuilder();
        sqlStr.append(" select kafkid, rwid, jdxx, jgbm, pyl, bm, nr, zdm, xxzt ");
        sqlStr.append("   from bdu_kafka_msg ");
        sqlStr.append("  where upper(bm) = upper('" + tableName + "') ");
        sqlStr.append("    and xxzt = '0' ");
        sqlStr.append("  order by cjsj asc ");
        try {
            //导出待处理记录的csv文件
            dataBaseService.copyTableData2Csv(filename, sqlStr.toString());
            //读取csv文件数据
            List<BduKafkaMsg> msgList = CsvUtils.load(filename, BduKafkaMsg.class);
            new File(filename).delete();
            return msgList;
        } catch (Exception e) {
            log.error("加载待处理数据记录失败！", e);
        }
        return Lists.newArrayList();
    }

    /**
     * 处理状态更新
     *
     * @param kafkid
     * @param result ：true成功、false失败
     */
    @Override
    public void updateStatus(String kafkid, boolean result) {
        bduKafkaMsgMapper.updateStatus(kafkid, result ? "1" : "2");
    }
}
